package com.demo.rocketmq.controller;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.demo.rocketmq.config.RocketMqDelayLevelEnum;
import com.demo.rocketmq.config.TopicConfig;
import com.demo.rocketmq.model.DemoOrder;
import com.demo.rocketmq.sender.RocketMqSender;
import com.demo.rocketmq.sender.TransactionMqSender;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;
import java.util.List;



/**
 * @author wangfengchen
 * 官方文档：<a href="https://rocketmq.apache.org/zh/docs/featureBehavior/02delaymessage">https://rocketmq.apache.org/zh/docs/featureBehavior/02delaymessage</a>
 */
@RestController
@Slf4j
@RequestMapping("/mq")
public class TestController {
    private static final Logger logger = LoggerFactory.getLogger(TestController.class);

    @Autowired
    private RocketMqSender rocketMqSender;
    @Autowired
    private TopicConfig topicConfig;
    @Autowired
    private TransactionMqSender transactionMqSender;


    /**
     * 发送同步消息
     *
     * @param msg
     * @return
     */
    @GetMapping("/sendMq")
    public boolean sendMq(String msg) {
        SendResult sendResult = rocketMqSender.send(topicConfig.getDemoTopic(), "1", msg);
        if (ObjectUtils.isNotEmpty(sendResult) && sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送消息成功，sendResult = {}", sendResult);
            return true;
        } else {
            log.error("MQ发送消息失败，sendResult = {}", sendResult);
            return false;
        }
    }

    /**
     * 发送同步延时消息级别
     * @param msg
     * @return
     */
    @GetMapping("/sendDelayMq")
    public boolean sendDelayMq(String msg) {
        SendResult sendResult = rocketMqSender.sendDelay(topicConfig.getDemoTopic(), "2", msg, RocketMqDelayLevelEnum.time10s);
        if (ObjectUtils.isNotEmpty(sendResult) && sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送消息成功，sendResult = {}", sendResult);
            return true;
        } else {
            log.error("MQ发送消息失败，sendResult = {}", sendResult);
            return false;
        }
    }

    /**
     * 发送同步延时消息秒
     * @param msg
     * @return
     */
    @GetMapping("/sendDelayBySecMq")
    public boolean sendDelayBySec(String msg) {
        SendResult sendResult = rocketMqSender.sendDelayBySec(topicConfig.getDemoTopic(), "3", msg, 10);
        if (ObjectUtils.isNotEmpty(sendResult) && sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送消息成功，sendResult = {}", sendResult);
            return true;
        } else {
            log.error("MQ发送消息失败，sendResult = {}", sendResult);
            return false;
        }
    }

    /**
     * 发送同步延时消息毫秒
     * @param msg
     * @return
     */
    @GetMapping("/sendDelayByMsMq")
    public boolean sendDelayByMs(String msg) {
        SendResult sendResult = rocketMqSender.sendDelayByMs(topicConfig.getDemoTopic(), "3", msg, 10000);
        if (ObjectUtils.isNotEmpty(sendResult) && sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送消息成功，sendResult = {}", sendResult);
            return true;
        } else {
            log.error("MQ发送消息失败，sendResult = {}", sendResult);
            return false;
        }
    }

    /**
     * 发送同步定时消息
     * @param msg
     * @return
     */
    @GetMapping("/sendDelayByDeliverTimeMsMq")
    public boolean sendDelayByDeliverTimeMs(String msg) {
        long deliverTime = DateUtil.offsetSecond(new Date(), 10).getTime();
//        long deliverTime = System.currentTimeMillis() + 10 * 1000;
        SendResult sendResult = rocketMqSender.sendDelayByDeliverTimeMs(topicConfig.getDemoTopic(), "4", msg, deliverTime);
        if (ObjectUtils.isNotEmpty(sendResult) && sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送消息成功，sendResult = {}", sendResult);
            return true;
        } else {
            log.error("MQ发送消息失败，sendResult = {}", sendResult);
            return false;
        }
    }

    /**
     * 发送同步顺序消息
     * @return
     */
    @GetMapping("/sendSequenceMq")
    public boolean sendSequenceMq() {
        List<DemoOrder> list = DemoOrder.getOrderList();
        for (DemoOrder demoOrder : list) {
            SendResult sendResult = rocketMqSender.sendWithSelector(topicConfig.getSequenceTopic(), String.valueOf(demoOrder.getOrderId()), JSON.toJSONString(demoOrder), String.valueOf(demoOrder.getOrderId()));
            if (ObjectUtils.isNotEmpty(sendResult) && sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
                log.info("MQ发送消息成功，sendResult = {}", sendResult);
            } else {
                log.error("MQ发送消息失败，sendResult = {}", sendResult);
                return false;
            }
        }
        return true;
    }

    /**
     * 异步发送消息
     *
     * @param msg
     * @return
     */
    @GetMapping("/sendAsyncMq")
    public boolean sendAsyncMq(String msg) {
        rocketMqSender.sendAsync(topicConfig.getDemoTopic(), "4", msg);
        return true;
    }

    /**
     * 异步延时消息
     *
     * @param msg
     * @return
     */
    @GetMapping("/sendAsyncDelayMq")
    public boolean sendAsyncDelayMq(String msg) {
        rocketMqSender.sendAsyncDelay(topicConfig.getDemoTopic(), "5", msg, RocketMqDelayLevelEnum.time10s);
        return true;
    }

    /**
     * 异步顺序消息
     * @return
     */
    @GetMapping("/sendAsyncSequenceMq")
    public boolean sendAsyncSequenceMq() {
        List<DemoOrder> list = DemoOrder.getOrderList();
        for (DemoOrder demoOrder : list) {
            rocketMqSender.sendAsyncWithSelector(topicConfig.getSequenceTopic(), String.valueOf(demoOrder.getOrderId()), JSON.toJSONString(demoOrder), String.valueOf(demoOrder.getOrderId()));
        }
        return true;
    }

    /**
     * 事务发送消息
     * @param msg
     * @return
     */
    @RequestMapping("/sendTransactionMq")
    public boolean sendTransactionMq(String msg) {
        try {
            SendResult sendResult = transactionMqSender.sendMsg(topicConfig.getDemoTopic(), msg, null, "7");
            if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
                log.info("MQ发送事务消息成功，sendResult = {}", sendResult);
                return true;
            } else {
                log.error("MQ发送事务消息失败，sendResult = {}", sendResult);
                return false;
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("MQ发送事务消息异常，e = {}", ExceptionUtils.getStackTrace(e));
            return false;
        }
    }

}
